package indi.lqzhi.tcpserver.pipeline;


import indi.lqzhi.tcpserver.message.BusMessage;
import lombok.extern.slf4j.Slf4j;
import lombok.var;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;


/**
 * @Author: LQZhi
 * @ClassName: BusPipeline
 * @Date: 2021-09-30 00-12
 * @Description: This is a BusPipeline Class.
 * @Version: V1.0
 */
@Slf4j
public class BusPipeline extends AbstractPipeline<BusMessage>{

    /**
     * 总线调度线程池
     */
    private final Executor asyncBusExecutor;

    /**
     * 读取消息线程的停止位
     */
    private volatile boolean readRunnableStopFlag;

    /**
     * 写入消息线程的停止位
     */
    private volatile boolean writeRunnableStopFlag;

    private final Runnable readRunnable = () -> {
        log.debug("{}总线：启动了读取线程",Thread.currentThread().getName());
        while (!readRunnableStopFlag){
            try {
                //尝试从读取的消息队列中取出数据
                var message = readMessageDeque.take();
                if ("".equals(message.getId()) && message.getObject() == null){
                    continue;//继续，所有要在发送之前，将writeRunnableStopFlag设置为true，让其关闭
                }
                long readCountValue = incrementReadCount();//读事件加1
                readCountHandlers.forEach(consumer -> consumer.accept(readCountValue));//读指针事件分发
                //存在数据,分发读取的消息
                final BusMessage newValue = disposeReadMessage(message);//处理读事件
                readMessageHandlers.forEach(v -> v.accept(newValue));
            }catch (Exception exception){
                //抛出的异常欸
                log.error("{}总线：读取线程发送了异常 -> {}",Thread.currentThread().getName(),exception);
                readMessageExceptionHandlers.forEach(v -> v.accept(exception));
            }
        }
        readMessageDeque.clear();//清空，不再持有对象
        log.debug("{}总线：关闭了读取线程",Thread.currentThread().getName());
    };


    private final Runnable writeRunnable = () -> {
        log.debug("{}总线：启动了写入线程",Thread.currentThread().getName());
        while (!writeRunnableStopFlag){
            try {
                //尝试从写入的消息队列中取出数据
                var message = writeMessageDeque.take();
                if ("".equals(message.getId()) && message.getObject() == null){
                    continue;//继续，所有要在发送之前，将writeRunnableStopFlag设置为true，让其关闭
                }
                long writeCountValue = incrementWriteCount();
                //回调处理
                writeCountHandlers.forEach(v -> v.accept(writeCountValue));
                //存在数据,发送数据给所有的客户端的消息
                final var newValue = disposeWriteMessage(message);
                writeMessageHandlers.forEach(v -> v.accept(newValue));
            }catch (Exception exception){
                //抛出的异常欸
                log.error("{}总线：写入线程发送了异常 -> {}",Thread.currentThread().getName(),exception);
                writeMessageExceptionHandlers.forEach(v -> v.accept(exception));
            }
        }
        writeMessageDeque.clear();//清空，不再持有对象
        log.debug("{}总线：关闭了写入线程",Thread.currentThread().getName());
    };

    /**
     * 构造方法
     * @param asyncBusExecutor 线程池
     */
    public BusPipeline(Executor asyncBusExecutor) {
        super();
        this.asyncBusExecutor = asyncBusExecutor == null ? Executors.newFixedThreadPool(2) : asyncBusExecutor;
    }

    /**
     * 构造方法
     */
    public BusPipeline() {
        this(Executors.newFixedThreadPool(2));
    }


    /**
     * 开启总线
     */
    @Override
    public void start() {
        if (hasRuning){
            return;//如果是运行状态，则不管
        }
        asyncBusExecutor.execute(readRunnable);
        asyncBusExecutor.execute(writeRunnable);
        hasRuning = true;//启动
    }

    /**
     * 销毁总线
     */
    @Override
    public void destory() {
        readRunnableStopFlag = true;//停止线程
        writeRunnableStopFlag = true;//停止线程
        //发送空数据，等待线程停止
        BusMessage stopMessage = BusMessage.of("", null);
        sendReadMessage(stopMessage);
        sendWriteMessage(stopMessage);
        hasRuning = false;//停止
    }

}
